package main.java.black;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseWindowedBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.windowing.TupleWindow;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * TopIpCountBolt
 *
 * @author zhangyimin
 * @version 1.0
 * @date 2018-11-13 下午3:01
 */
public class BlackListCountByWindowBolt extends BaseWindowedBolt {

    private OutputCollector collector;

    //定义一个集合,保存窗口处理的结果
    private Map<String, Long> result = new HashMap<>();

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("user_id", "pv"));
    }

    @Override
    public void execute(TupleWindow tupleWindow) {

        List<Tuple> tuples = tupleWindow.get();
        for (Tuple tuple : tuples) {
            String userId = tuple.getStringByField("user_id");
            Long count = tuple.getLongByField("count");
            if (result.containsKey(userId)) {
                result.put(userId, result.get(userId) + count);
            } else {
                result.put(userId, count);
            }

            System.out.println("统计结果为:" + result);
            if (result.get(userId) > 4) {
                //大于4的次数 保存到数据库中
                collector.emit(new Values(userId,result.get(userId)));
            }

        }


    }
}
